package com.huanzhidadi.rabbitmq.seqqueuewithredis.consumer;

import com.alibaba.fastjson2.JSON;
import com.huanzhidadi.rabbitmq.seqqueuewithredis.constant.RedisConstants;
import com.huanzhidadi.rabbitmq.seqqueuewithredis.entity.MessageInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class ConsumerBase {

    @Autowired
    private RedisTemplate redisTemplate;


    void operMessage(MessageInfo message) {
        log.info(JSON.toJSONString(message));
        Integer preId = message.getPreId();
        log.info("preId: {}", preId);
        if (Objects.isNull(preId)) {
            // 如果前一id为null，说明是第一个元素，直接消费，加入batchNo为key的消费记录列表consumedList
            log.info("preId为空直接消费");
            consumMessage(message);
            // 检查后面节点
            checkPostMessage(message);
        } else {
            // 如果前一id不为null，从batchNo为key的消费记录列表consumedList中取出最后一个元素，判断id是否是前一id
            String consumedListKey = RedisConstants.CONSUMED_LIST + message.getBatchNo();
            String beforeConsumMapKey = RedisConstants.BEFORE_CONSUM_MAP + message.getBatchNo();

            // 消费记录列表consumedList中最后一个元素
            log.info("{}: rightPop前大小{}", consumedListKey, redisTemplate.opsForList().size(consumedListKey));
            if (0 == redisTemplate.opsForList().size(consumedListKey)) {
                // 如果不能获取，说明还没有接收到前一id的消息，加入beforeConsumMap
                log.info("不可以获取，加入beforeConsumMap: {}", JSON.toJSONString(message));
                redisTemplate.opsForHash().put(beforeConsumMapKey, String.valueOf(message.getId()), JSON.toJSONString(message));
                redisTemplate.expire(beforeConsumMapKey, RedisConstants.CONSUME_EXPIRE_TIME, TimeUnit.MILLISECONDS);
                return;
            }
            String lastConsumedMsgStr = (String) redisTemplate.opsForList().rightPop(consumedListKey);
            log.info("{}: rightPop后大小{}", consumedListKey, redisTemplate.opsForList().size(consumedListKey));
            log.info("lastConsumedMsgStr: {}", lastConsumedMsgStr);

            MessageInfo lastConsumedMsg = JSON.parseObject(lastConsumedMsgStr, MessageInfo.class);
            // 前面弹出list的元素还要再放回去
            redisTemplate.opsForList().rightPush(consumedListKey, lastConsumedMsgStr);
            log.info("{}: rightPush后大小{}", consumedListKey, redisTemplate.opsForList().size(consumedListKey));
            if (preId.equals(lastConsumedMsg.getId())) {
                log.info("id是前一id: {}", preId);
                // 如果是，直接消费，加入consumedList
                consumMessage(message);
                // 检查后面节点
                checkPostMessage(message);
            } else {
                // 如果不是，从beforeConsumSet中获取前一id的元素
                Boolean hasKey = redisTemplate.opsForHash().hasKey(beforeConsumMapKey, String.valueOf(preId));
                log.info("id不是前一id，查询beforeConsumMap是否存在: {}", hasKey);
                if (true == hasKey) {
                    // 如果可以获取，取出前一id的元素
                    String beforeConsumMsgStr = (String) redisTemplate.opsForHash().get(beforeConsumMapKey, String.valueOf(preId));
                    log.info("可以获取，取出前一id的元素: {}", beforeConsumMsgStr);
                    MessageInfo beforeConsumMsg = JSON.parseObject(beforeConsumMsgStr, MessageInfo.class);
                    consumMessage(beforeConsumMsg);
                    consumMessage(message);
                    // 检查后面节点
                    checkPostMessage(message);
                } else {
                    // 如果不能获取，说明还没有接收到前一id的消息，加入beforeConsumMap
                    log.info("不可以获取，加入beforeConsumMap: {}", JSON.toJSONString(message));
                    redisTemplate.opsForHash().put(beforeConsumMapKey, String.valueOf(message.getId()), JSON.toJSONString(message));
                    redisTemplate.expire(beforeConsumMapKey, RedisConstants.CONSUME_EXPIRE_TIME, TimeUnit.MILLISECONDS);
                }
            }
        }
    }

    /**
     * 消费当前消息
     * @param message
     */
    private void consumMessage(MessageInfo message) {
        log.info(message.getId() + ":" + message.getMsg());
        String consumedListKey = RedisConstants.CONSUMED_LIST + message.getBatchNo();
        String beforeConsumMapKey = RedisConstants.BEFORE_CONSUM_MAP + message.getBatchNo();

        log.info("message.getPostId(): {}", message.getPostId());
        // 如果是最后一个节点，删除redis缓存
        if (null == message.getPostId()) {
            log.info("最后一个节点，删除redis缓存");
            redisTemplate.delete(consumedListKey);
            redisTemplate.delete(beforeConsumMapKey);
        } else {
            log.info("加入consumedList：{}", JSON.toJSONString(message));
            log.info("{}: rightPush前大小{}", consumedListKey, redisTemplate.opsForList().size(consumedListKey));
            redisTemplate.opsForList().rightPush(consumedListKey, JSON.toJSONString(message));
            log.info("{}:rightPush后大小{}", consumedListKey, redisTemplate.opsForList().size(consumedListKey));
            redisTemplate.expire(consumedListKey, RedisConstants.CONSUME_EXPIRE_TIME, TimeUnit.MILLISECONDS);
        }
    }

    /**
     * 检查后面节点
     * @param message
     */
    private void checkPostMessage(MessageInfo message) {
        Integer postId = message.getPostId();
        String beforeConsumMapKey = RedisConstants.BEFORE_CONSUM_MAP + message.getBatchNo();
        Boolean hasKey = redisTemplate.opsForHash().hasKey(beforeConsumMapKey, String.valueOf(postId));
        log.info("检查后面节点{}，查询beforeConsumMap是否存在: {}", postId, hasKey);
        if (true == hasKey) {
            // 如果可以获取，取出检查后面节点的元素
            String beforeConsumMsgStr = (String) redisTemplate.opsForHash().get(beforeConsumMapKey, String.valueOf(postId));
            log.info("可以获取，取出检查后面节点的元素: {}", beforeConsumMsgStr);
            MessageInfo beforeConsumMsg = JSON.parseObject(beforeConsumMsgStr, MessageInfo.class);
            consumMessage(beforeConsumMsg);
        }
    }
}
